package cn.learn.server;

import cn.learn.proto.CreateRequest;
import cn.learn.server.ZooKeeperServer.ChangeRecord;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class PrepRequestProcessor extends Thread implements RequestProcessor {
    ZooKeeperServer zks;
    private final RequestProcessor nextProcessor;
    LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();

    public PrepRequestProcessor(ZooKeeperServer zks, RequestProcessor nextProcessor) {
        this.nextProcessor = nextProcessor;
        this.zks = zks;
    }

    @Override
    public void run() {
        try {
            while (true) {
                System.out.println("启动前置处理器线程");
                System.out.println(submittedRequests.size());
                Request request = submittedRequests.take();
                System.out.println(request);
                pRequest(request);
            }
        }catch (Exception e) {

        }
    }

    protected void pRequest(Request request) throws RequestProcessorException {
//        request.setHdr(null);
//        request.setTxn(null);
        pRequestHelper(request);
        request.zxid = zks.getZxid();
        nextProcessor.processRequest(request);
    }

    private void pRequestHelper(Request request) throws RequestProcessorException {
        try {
            switch (request.type) {
                case 1: //create
                    CreateRequest create2Request = new CreateRequest();
                    pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
                    break;
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected void pRequest2Txn(int type, long zxid, Request request, Record record, boolean deserialize) throws IOException, RequestProcessorException {
        switch (type) {
            case 1:
                pRequest2TxnCreate(type, request, record, deserialize);
                break;
        }
    }

    // 创建一个新节点
    private void pRequest2TxnCreate(int type, Request request, Record record, boolean deserialize) throws IOException {
        if (deserialize) {
            ByteBufferInputStream.byteBuffer2Record(request.request, record);
        }
        String path;
        byte[] data;
        CreateRequest createRequest = (CreateRequest) record;
        path = createRequest.getPath();
        data = createRequest.getData();
        String parentPath = validatePathForCreate(path, request.sessionId);
        // 查该路径是否存在
        ChangeRecord parentRecord = getRecordForPath(parentPath);
//        parentRecord = parentRecord.duplicate(request);
        parentRecord.childCount++;
        // 修改父节点
        addChangeRecord(parentRecord);
        // 添加子节点
        ChangeRecord nodeRecord = new ChangeRecord(1, path, 0);
        nodeRecord.data = data;
        addChangeRecord(nodeRecord);
    }

    protected void addChangeRecord(ChangeRecord c) {
        synchronized (zks.outstandingChanges) {
            zks.outstandingChanges.add(c);
            zks.outstandingChangesForPath.put(c.path, c);
        }
    }

    private ChangeRecord getRecordForPath(String path){
        ChangeRecord lastChange = null;
        synchronized (zks.outstandingChanges) {
            lastChange = zks.outstandingChangesForPath.get(path);
            if (lastChange == null) {
                DataNode n = zks.getZKDatabase().getNode(path);
                if (n != null) {
                    Set<String> children;
                    synchronized (n) {
                        children = n.getChildren();
                    }
                    lastChange = new ChangeRecord(-1, path, children.size());
                    lastChange.data = n.getData();
                }
            }
        }
        return lastChange;
    }

    private String validatePathForCreate(String path, long sessionId) {
        int lastSlash = path.lastIndexOf('/');
        return path.substring(0, lastSlash);
    }

    @Override
    public void processRequest(Request request) throws RequestProcessorException {
        System.out.println("process处理器开始处理：校验数据，将数据放到队列中等待执行");
        submittedRequests.add(request);
        System.out.println(submittedRequests.size());
    }

    @Override
    public void shutdown() {

    }
}
